package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.OrderDetail;
import com.atguigu.gmall.realtime.beans.OrderInfo;
import com.atguigu.gmall.realtime.beans.OrderWide;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * Author: Felix
 * Date: 2021/11/30
 * Desc: 订单宽表的准备
 *  -订单和订单明细双流join
 *  -订单和订单明细双流join的结果和维度进行关联
 *需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、Redis、BaseDBApp、OrderWideApp
 *执行流程
 *      -运行模拟生成业务数据的jar
 *      -将生成的业务数据保存到业务数据库表中
 *      -binlog会将业务数据库表的变化记录下来
 *      -maxwell从binlog中读取变化并封装为json字符串发给kafka的ods_base_db_m主题
 *      -BaseDBApp应用从ods_base_db_m主题中读取数据进行分流
 *          >事实数据  写到kafka不同的主题中
 *          >维度数据  写到phoenix维度表汇总
 *              在测试之前，先将配置表的信息完善，提前创建维度表
 *              在测试之前，使用maxwell的bootstrap对维度历史数据进行同步
 *      -OrderWideApp从dwd_order_info和dwd_order_detail主题中读取数据，双流join
 *      -和用户维度进行管理
 *          PhoenixUtil->DimUtil getDimInfoNoCache->DimUtil getDimInfo 旁路缓存->异步IO
 *      -异步IO
 *          >将异步操作应用到流上
 *              AsyncDataStream.unorderWait(
 *                  流,
 *                  实现AsyncFunction异步操作,
 *                  超时时间,
 *                  时间单位
 *              )
 *         >封装DimAsyncFunction extends RichAsyncFunction{
 *             asyncInvoke(){
 *                  开启多个线程，发送异步请求
 *                  使用模板方法设计模式，在这个类中定义实现维度关联的功能的核心算法的骨架，具体的实现我们交个子类去做。
 *             }
 *         }
 */
public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关设置(略)
        //TODO 3.从kafka中读取数据
        //3.1 声明消费的主题以及消费者组
        String orderInfoTopic = "dwd_order_info";
        String orderDetailTopic = "dwd_order_detail";
        String groupId = "order_wide_app_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> orderInfoKafkaSource = MyKafkaUtil.getKafkaSource(orderInfoTopic, groupId);
        FlinkKafkaConsumer<String> orderDetailKafkaSource = MyKafkaUtil.getKafkaSource(orderDetailTopic, groupId);

        //3.3 消费数据 封装为流
        DataStreamSource<String> orderInfoStrDS = env.addSource(orderInfoKafkaSource);
        DataStreamSource<String> orderDetailStrDS = env.addSource(orderDetailKafkaSource);

        //TODO 4.对读取的数据进行类型的转换  jsonStr--->实体类型对象
        //4.1 订单
        SingleOutputStreamOperator<OrderInfo> orderInfoDS = orderInfoStrDS.map(
                new RichMapFunction<String, OrderInfo>() {
                    private  SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderInfo map(String jsonStr) throws Exception {
                        OrderInfo orderInfo = JSON.parseObject(jsonStr, OrderInfo.class);
                        orderInfo.setCreate_ts(sdf.parse(orderInfo.getCreate_time()).getTime());
                        return orderInfo;
                    }
                }
        );
        //4.2 订单明细
        SingleOutputStreamOperator<OrderDetail> orderDetailDS = orderDetailStrDS.map(
                new RichMapFunction<String, OrderDetail>() {
                    private SimpleDateFormat sdf ;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderDetail map(String jsonStr) throws Exception {
                        OrderDetail orderDetail = JSON.parseObject(jsonStr, OrderDetail.class);
                        orderDetail.setCreate_ts(sdf.parse(orderDetail.getCreate_time()).getTime());
                        return orderDetail;
                    }
                }
        );

        //orderInfoDS.print(">>>>");
        //orderDetailDS.print("####");

        //TODO 5.指定Watermark以及提取事件时间字段
        //5.1 订单流
        SingleOutputStreamOperator<OrderInfo> orderInfoWithWatermarkDS = orderInfoDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<OrderInfo>() {
                                    @Override
                                    public long extractTimestamp(OrderInfo orderInfo, long recordTimestamp) {
                                        return orderInfo.getCreate_ts();
                                    }
                                }
                        )
        );
        //5.2 订单明细流
        SingleOutputStreamOperator<OrderDetail> orderDetailWithWatermarkDS = orderDetailDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<OrderDetail>() {
                                    @Override
                                    public long extractTimestamp(OrderDetail orderDetail, long recordTimestamp) {
                                        return orderDetail.getCreate_ts();
                                    }
                                }
                        )
        );

        //TODO 6.通过keyby对两条流进行分组  指定连接字段
        //6.1 订单
        KeyedStream<OrderInfo, Long> orderInfoKeyedDS = orderInfoWithWatermarkDS.keyBy(OrderInfo::getId);
        //6.2 订单明细
        KeyedStream<OrderDetail, Long> orderDetailKeyedDS = orderDetailWithWatermarkDS.keyBy(OrderDetail::getOrder_id);

        //TODO 7.双流join  使用intervalJoin将订单和订单明细连接在一起
        SingleOutputStreamOperator<OrderWide> joinedDS = orderInfoKeyedDS
                .intervalJoin(orderDetailKeyedDS)
                .between(Time.seconds(-5), Time.seconds(5))
                .process(
                        new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                            @Override
                            public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, Context ctx, Collector<OrderWide> out) throws Exception {
                                out.collect(new OrderWide(orderInfo, orderDetail));
                            }
                        }
                );

        //joinedDS.print(">>>>");

        //TODO 8.和用户维度进行关联
        /*joinedDS.map(
            new MapFunction<OrderWide, OrderWide>() {
                @Override
                public OrderWide map(OrderWide orderWide) throws Exception {
                    //获取维度的id
                    Long userId = orderWide.getUser_id();
                    //根据id到维度表中查询维度数据
                    JSONObject dimInfoJsonObj = DimUtil.getDimInfo("DIM_USER_INFO", userId.toString());
                    //将查询出来的维度数据  赋值给对象的属性
                    String gender = dimInfoJsonObj.getString("GENDER");
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                    String birthdayDate = dimInfoJsonObj.getString("BIRTHDAY");
                    Long birthdayTime = sdf.parse(birthdayDate).getTime();
                    Long curTime = System.currentTimeMillis();
                    Long ageLong = (curTime - birthdayTime) / 1000L / 60L / 60L / 24L / 365L;
                    orderWide.setUser_gender(gender);
                    orderWide.setUser_age(ageLong.intValue());
                    return orderWide;
                }
            }
        );*/
        //将异步I/O操作应用于DataStream作为DataStream的一次转换操作
        SingleOutputStreamOperator<OrderWide> orderWidewithUserInfoDS = AsyncDataStream.unorderedWait(
                joinedDS,
                //异步操作需要实现分发请求的 AsyncFunction
                new DimAsyncFunction<OrderWide>("DIM_USER_INFO") {
                    @Override
                    public void join(JSONObject dimInfoJsonObj, OrderWide orderWide) throws Exception {
                        String gender = dimInfoJsonObj.getString("GENDER");

                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                        String birthdayDate = dimInfoJsonObj.getString("BIRTHDAY");

                        Long birthdayTime = sdf.parse(birthdayDate).getTime();
                        Long curTime = System.currentTimeMillis();
                        Long ageLong = (curTime - birthdayTime) / 1000L / 60L / 60L / 24L / 365L;
                        orderWide.setUser_gender(gender);
                        orderWide.setUser_age(ageLong.intValue());
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getUser_id().toString();
                    }
                },
                60, TimeUnit.SECONDS
        );
        //orderWidewithUserInfoDS.print(">>>>");

        //TODO 9.和地区维度进行关联
        SingleOutputStreamOperator<OrderWide> orderWideWithProvinceDS = AsyncDataStream.unorderedWait(
                orderWidewithUserInfoDS,
                new DimAsyncFunction<OrderWide>("DIM_BASE_PROVINCE") {
                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getProvince_id().toString();
                    }

                    @Override
                    public void join(JSONObject provinceDimInfo, OrderWide orderWide) throws Exception {
                        orderWide.setProvince_name(provinceDimInfo.getString("NAME"));
                        orderWide.setProvince_area_code(provinceDimInfo.getString("AREA_CODE"));
                        orderWide.setProvince_iso_code(provinceDimInfo.getString("ISO_CODE"));
                        orderWide.setProvince_3166_2_code(provinceDimInfo.getString("ISO_3166_2"));
                    }
                },
                60, TimeUnit.SECONDS
        );
        //orderWideWithProvinceDS.print(">>>>>");

        //TODO 10.和SKU维度进行关联
        SingleOutputStreamOperator<OrderWide> orderWideWithSkuInfoDS = AsyncDataStream.unorderedWait(
                orderWideWithProvinceDS,
                new DimAsyncFunction<OrderWide>("DIM_SKU_INFO") {
                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getSku_id().toString();
                    }

                    @Override
                    public void join(JSONObject skuDimInfo, OrderWide orderWide) throws Exception {
                        orderWide.setSku_name(skuDimInfo.getString("SKU_NAME"));
                        orderWide.setTm_id(skuDimInfo.getLong("TM_ID"));
                        orderWide.setCategory3_id(skuDimInfo.getLong("CATEGORY3_ID"));
                        orderWide.setSpu_id(skuDimInfo.getLong("SPU_ID"));
                    }
                }, 60, TimeUnit.SECONDS
        );
        //orderWideWithSkuInfoDS.print(">>>>");

        //TODO 11.和SPU维度进行关联
        SingleOutputStreamOperator<OrderWide> orderWideWithSpuDS = AsyncDataStream.unorderedWait(
                orderWideWithSkuInfoDS,
                new DimAsyncFunction<OrderWide>("DIM_SPU_INFO") {
                    @Override
                    public void join(JSONObject jsonObject,OrderWide orderWide) throws Exception {
                        orderWide.setSpu_name(jsonObject.getString("SPU_NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getSpu_id());
                    }
                }, 60, TimeUnit.SECONDS);

        //TODO 12.和品类维度进行关联
        SingleOutputStreamOperator<OrderWide> orderWideWithCategory3DS = AsyncDataStream.unorderedWait(
                orderWideWithSpuDS,
                new DimAsyncFunction<OrderWide>("DIM_BASE_CATEGORY3") {
                    @Override
                    public void join(JSONObject jsonObject,OrderWide orderWide) throws Exception {
                        orderWide.setCategory3_name(jsonObject.getString("NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getCategory3_id());
                    }
                }, 60, TimeUnit.SECONDS);


        //TODO 13.和品牌维度进行关联
        SingleOutputStreamOperator<OrderWide> orderWideWithTmDS = AsyncDataStream.unorderedWait(
                orderWideWithCategory3DS,
                new DimAsyncFunction<OrderWide>("DIM_BASE_TRADEMARK") {
                    @Override
                    public void join(JSONObject jsonObject,OrderWide orderWide) throws Exception {
                        orderWide.setTm_name(jsonObject.getString("TM_NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getTm_id());
                    }
                }, 60, TimeUnit.SECONDS);

        orderWideWithTmDS.print(">>>>");

        //TODO 14.将完整的订单宽表数据写回到kafka的dwm层
        orderWideWithTmDS
                .map(orderWide->JSON.toJSONString(orderWide))
                .addSink(MyKafkaUtil.getKafkaSink("dwm_order_wide"));

        env.execute();
    }
}
